package KafkaTestProducer;



import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.util.Properties;


/**
 * Created by Ljc on 5/4/2017.
 */


public class KafkaTestProducer {
	
	public static Producer<String,String> init(){
		
        // create instance for properties to access producer configs
        Properties props = new Properties();

        //Assign localhost id
        //props.put("bootstrap.servers", "student27-x1:9092,student27-x2:9092,student28-x1:9092,student28-x2:9092");
        props.put("bootstrap.servers", "student39-x1:9092,student39-x2:9092,student40-x1:9092,student40-x2:9092");
        
        //Set acknowledgements for producer requests.
        props.put("acks", "all");

        //If the request fails, the producer can automatically retry,
        props.put("retries", 0);

        //Specify buffer size in config
        props.put("batch.size", 16384);

        //Reduce the no of requests less than 0
        props.put("linger.ms", 1);

        //The buffer.memory controls the total amount of memory available to the producer for buffering.
        props.put("buffer.memory", 33554432);

        props.put("key.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");

        props.put("value.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new KafkaProducer<String, String>(props);
        return producer;
	}
	
    public static void main(String[] args) throws Exception{


    	  String topicName1 = "sparkfinal4";
    	  
    	  Producer<String, String> producer = init();

            //read a big file(contain lots of sentence) 
            File file = new File("/home/hduser/reader.txt");
            FileReader reader = new FileReader(file);
            BufferedReader bufferedReader = new BufferedReader(reader);
            String send = null;
            int index = 0;
            
            
            while((send = bufferedReader.readLine()) != null){
                //send  sentence
	            producer.send(new ProducerRecord<String, String>(topicName1, Integer.toString(index++), send));
	            if (index%10000 == 0)
	            	System.out.println("Message sending");
            }                
        System.out.println("Message sent successfully");
        bufferedReader.close();
        producer.close();
    }
}